#include "aio_tcp.h"
#include "../global/head.h"
#include "../xos/head.h"
#include "../buf/head.h"
#include "aio_callback_thread.h"
#include "../msg/head.h"
#include "aio_impl.h"
#include "aio_key.h"

//  if(errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
// 
/////////////////////////////////////////////////////////////////////////////////////////
// 
namespace xos
{
    aio_tcp::aio_tcp( aio_impl * pAioImpl ) : m_pAioImpl( pAioImpl )
    {
    }

    aio_tcp::~aio_tcp()
    {
    }

    /////////////////////////////////////////////////////////////////////////////////////
    // 

    // 
    // m_nData[0] : cmd type
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::close_handle( msg_impl * lpMsg )
    {
        using namespace std;
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->m_lpData[1];
        udp_impl * lpUdp = ( udp_impl * )lpKey->m_pHandle;
        socket_impl * lpSocket = &lpUdp->m_socket;
        int hSocket = lpSocket->m_hSocket;

        epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_DEL, hSocket, 0 );
        lpKey->finish_request( m_pAioImpl );
        lpKey->release();
        lpKey = 0;

        // 
        // return.
        // 
        lpMsg->m_nData[0] = i_aio::AIO_TCP_CLOSE_HANDLE_RET;
        lpMsg->m_bData[0] = true;

        m_pAioImpl->m_pCallbackThread->notify( lpMsg );
        lpMsg = 0;

        return nRet;
    }

    // 
    // m_bData[0] : true or false for result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::user_init_server( msg_impl * lpMsg )
    {
        using namespace std;
        int nRet = 0;

        socket_impl * lpSocket = 0;
        tcp_impl * lpTcp = 0;
        aio_key * lpKey = 0;

        lpMsg->m_nData[0] = i_aio::AIO_TCP_INIT_SERVER_RET;
        lpMsg->m_bData[0] = false;

        if( 0 == nRet )
        {
            nRet = g_pXosImpl->create( i_xos::XOS_OBJ_TCP, ( void ** )&lpTcp );
        }

        if( 0 == nRet )
        {
            lpSocket = &lpTcp->m_socket;
            nRet = lpSocket->tcp_init( lpMsg->m_szStr[1], lpMsg->m_nData[1] );
            if( 0 == nRet )
            {
                lpSocket->set_async( true );
            }
        }

        if( 0 == nRet )
        {
            lpKey = m_pAioImpl->m_key_pool.GetItem();
            lpKey->init();
            lpKey->m_nEpollFlags = EPOLLIN | EPOLLERR;
            lpKey->m_lpUserKey = lpMsg->m_lpData[0];
            lpMsg->m_lpData[1] = lpKey;
            lpKey->m_enumHandleType = aio_key::H_TYPE_TCP_SERVER;
            lpKey->m_pHandle = lpTcp;
            lpKey->m_eState = aio_key::KEY_STATE_RUNNING;
        }

        if( 0 == nRet )
        {
            lpMsg->m_bData[0] = true;
            struct epoll_event ev;
            ev.events = lpKey->m_nEpollFlags;
            ev.data.ptr = lpKey;
            epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_ADD, lpSocket->m_hSocket, &ev );
        }
        else
        {
            lpMsg->m_bData[0] = false;
        }

        m_pAioImpl->m_pCallbackThread->notify( lpMsg );

        if( 0 == nRet )
        {
            lpMsg = 0;
            lpTcp = 0;
            lpKey = 0;
        }

        if( lpKey )
        {
            lpKey->m_pHandle = 0;
            lpKey->release();
            lpKey = 0;
        }

        if( lpTcp )
        {
            lpTcp->release();
            lpTcp = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // m_lpData[2] : ClientKey
    // 
    int aio_tcp::user_accept( msg_impl * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->m_lpData[1];
        msg_impl * lpmsg = 0;

        lpMsg->m_nData[0] = i_aio::AIO_TCP_ACCPET_RET;
        lpMsg->m_bData[0] = false;

        if( lpKey->m_accept_list.size() > 0 )
        {
            lpmsg = lpKey->m_accept_list.front();
            lpKey->m_accept_list.pop_front();
        }

        if( lpmsg )
        {
            strcpy( lpMsg->m_szStr[1], lpmsg->m_szStr[1] );
            lpMsg->m_nData[1] = lpmsg->m_nData[1];
            strcpy( lpMsg->m_szStr[2], lpmsg->m_szStr[2] );
            lpMsg->m_nData[2] = lpmsg->m_nData[2];
            lpMsg->m_bData[0] = lpmsg->m_bData[0];
            lpMsg->m_nData[0] = lpmsg->m_nData[0];
            lpMsg->m_lpData[1] = lpmsg->m_lpData[1];
            lpMsg->m_lpData[2] = lpmsg->m_lpData[2];

            lpmsg->release();
            lpmsg = 0;

            m_pAioImpl->m_pCallbackThread->notify( lpMsg );
            lpMsg = 0;
        }
        else
        {
            lpKey->m_pend_accept_list.push_back( lpMsg );
            lpMsg = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::user_connect( msg_impl * lpMsg )
    {
        int nRet = 0;

        unsigned long dwIp = 0;
        char szIp[100] = { 0 };

        socket_impl * lpSocket = 0;
        tcp_impl * lpTcp = 0;
        aio_key * lpKey = 0;
        int hSocket = -1;

        lpMsg->m_nData[0] = i_aio::AIO_TCP_CONNECT_RET;
        lpMsg->m_bData[0] = false;

        if( 0 == nRet )
        {
            nRet = g_pXosImpl->create( i_xos::XOS_OBJ_TCP, ( void ** )&lpTcp );
        }

        if( 0 == nRet )
        {
            lpSocket = &lpTcp->m_socket;
            nRet = lpSocket->tcp_init( lpMsg->m_szStr[1], lpMsg->m_nData[1] );
            if( 0 == nRet )
            {
                hSocket = lpSocket->m_hSocket;
                lpSocket->set_async( true );
            }
        }

        if( 0 == nRet )
        {
            lpKey = m_pAioImpl->m_key_pool.GetItem();
            lpKey->init();
            lpKey->m_nEpollFlags = EPOLLIN | EPOLLOUT | EPOLLERR;
            lpKey->m_lpUserKey = lpMsg->m_lpData[0];
            lpKey->m_enumHandleType = aio_key::H_TYPE_TCP_CONN;
            lpKey->m_pHandle = lpTcp;
            lpKey->m_lpMsg = lpMsg;
            lpMsg->m_lpData[1] = lpKey;
        }

        if( 0 == nRet )
        {
            struct epoll_event ev;
            ev.events = lpKey->m_nEpollFlags;
            ev.data.ptr = lpKey;
            epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_ADD, hSocket, &ev );

            lpSocket->get_ip_by_name( lpMsg->m_szStr[2], dwIp, szIp );
            sockaddr_in addr = { 0 };
            addr.sin_family = AF_INET;
            addr.sin_addr.s_addr = dwIp;
            addr.sin_port = htons( lpMsg->m_nData[2] );

            connect( hSocket, ( const sockaddr * )&addr, sizeof( sockaddr_in ) );
            lpKey = 0;
            lpTcp = 0;
            lpMsg = 0;
        }

        if( 0 != nRet )
        {
            m_pAioImpl->m_pCallbackThread->notify( lpMsg );
            lpMsg = 0;
        }

        if( lpKey )
        {
            lpKey->m_pHandle = 0;
            lpKey->release();
            lpKey = 0;
        }

        if( lpTcp )
        {
            lpTcp->release();
            lpTcp = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : send bytes
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::user_send( msg_impl * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->m_lpData[1];

        lpMsg->m_nData[0] = i_aio::AIO_TCP_SEND_RET;
        lpMsg->m_bData[0] = false;

        lpKey->m_pend_send_list.push_back( lpMsg );
        lpMsg = 0;

        if( aio_key::KEY_STATE_RUNNING != lpKey->m_eState )
        {
            lpKey->finish_request( m_pAioImpl );
        }
        else
        {
            tcp_send( lpKey );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : recv len
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::user_recv( msg_impl * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->m_lpData[1];
        msg_impl * lpmsg = 0;
        buf_impl * lpBuf = lpMsg->m_lpBuf[0];

        lpMsg->m_nData[0] = i_aio::AIO_TCP_RECV_RET;
        lpMsg->m_bData[0] = false;

        if( lpKey->m_recv_list.size() > 0 )
        {
            lpmsg = lpKey->m_recv_list.front();
            lpKey->m_recv_list.pop_front();
        }

        if( lpmsg )
        {
            buf_impl *& lpbuf = lpmsg->m_lpBuf[0];

            lpMsg->m_bData[0] = lpmsg->m_bData[0];
            lpMsg->m_nData[0] = lpmsg->m_nData[0];
            lpMsg->m_nData[1] = lpmsg->m_nData[1];
            memcpy( lpBuf->m_szBuf, lpbuf->m_szBuf + lpbuf->m_nPos, lpbuf->m_nLen );
            lpBuf->m_nLen = lpbuf->m_nLen;
            lpMsg->m_lpData[1] = lpmsg->m_lpData[1];

            m_pAioImpl->m_pCallbackThread->notify( lpMsg );

            lpbuf->release();
            lpbuf = 0;

            lpmsg->release();
            lpmsg = 0;

            lpMsg = 0;
        }
        else
        {
            lpKey->m_pend_recv_list.push_back( lpMsg );
            lpMsg = 0;
        }

        if( aio_key::KEY_STATE_RUNNING != lpKey->m_eState )
        {
            lpKey->finish_request( m_pAioImpl );
        }

        return nRet;
    }

    ////////////////////////////////////////////////////////////////////////////////////////
    // 

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_recv_connect( aio_key * lpKey, int nFlags )
    {
        using namespace std;
        int nRet = 0;

        tcp_impl * lpTcp = ( tcp_impl * )lpKey->m_pHandle;
        socket_impl * lpSocket = &lpTcp->m_socket;
        msg_impl * lpMsg = lpKey->m_lpMsg;
        char szIp[100] = { 0 };
        xos_u16 usPort = 0;

        if( 0 == nRet )
        {
            if( !( nFlags & EPOLLOUT ) )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            nRet = lpSocket->get_peer_addr( szIp, &usPort );
        }

        if( 0 == nRet )
        {
            if( ( ( int )usPort != lpMsg->m_nData[2] ) || ( 0 != strcmp( szIp, lpMsg->m_szStr[2] ) ) )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            //nRet = lpSocket->get_socket_addr( szIp, &usPort );
        }

        if( 0 == nRet )
        {
            lpMsg->m_bData[0] = true;
            //strcpy( lpMsg->m_szStr[1], szIp );
            //lpMsg->m_nData[1] = usPort;

            lpKey->m_enumHandleType = aio_key::H_TYPE_TCP_NORM;
            lpKey->m_nEpollFlags &= ~EPOLLOUT;
            lpKey->m_bCanWrite = true;
            lpKey->m_lpMsg = 0;
            lpKey->m_eState = aio_key::KEY_STATE_RUNNING;

            struct epoll_event ev;
            ev.events = lpKey->m_nEpollFlags;
            ev.data.ptr = lpKey;
            epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_MOD, lpSocket->m_hSocket, &ev );
            lpKey = 0;
        }

        m_pAioImpl->m_pCallbackThread->notify( lpMsg );
        lpMsg = 0;

        if( lpKey )
        {
            lpKey->release();
            lpKey = 0;
        }

        return nRet;
    }

    //
    // tcp connect lost
    // 
    int aio_tcp::tcp_error( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        cout << "tcp_error" << endl;

        udp_impl * lpUdp = ( udp_impl * )lpKey->m_pHandle;
        socket_impl * lpSocket = &lpUdp->m_socket;
        int hSocket = lpSocket->m_hSocket;

        epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_DEL, hSocket, 0 );
        lpKey->finish_request( m_pAioImpl );

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // m_lpData[2] : ClientKey
    // 
    int aio_tcp::tcp_accept( aio_key * lpKey, int nFlags )
    {
        using namespace std;
        int nRet = 0;

        tcp_impl * lpTcp = ( tcp_impl * )lpKey->m_pHandle, * lpAcceptTcp = 0;
        socket_impl * lpSocket = &lpTcp->m_socket;
        msg_impl::LIST * lpList = 0;
        aio_key * lpRetKey = 0;
        msg_impl * lpMsg = 0;

        {
            lpRetKey = m_pAioImpl->m_key_pool.GetItem();
            lpRetKey->init();
        }

        if( 0 == lpKey->m_pend_accept_list.size() )
        {
            g_pXosImpl->create( i_xos::XOS_OBJ_MSG, ( void ** )&lpMsg );
            lpList = &lpKey->m_accept_list;
            lpMsg->m_lpData[1] = lpKey;
        }
        else
        {
            lpMsg = lpKey->m_pend_accept_list.front();
            lpKey->m_pend_accept_list.pop_front();
        }

        lpMsg->m_nData[0] = i_aio::AIO_TCP_ACCPET_RET;
        lpMsg->m_bData[0] = false;

        if( 0 == nRet )
        {
            nRet = lpSocket->tcp_accept( &lpAcceptTcp );
        }

        if( 0 == nRet )
        {
            char szIp[100] = { 0 };
            xos_u16 usPort = 0;
            lpAcceptTcp->m_socket.get_peer_addr( szIp, &usPort );
            strcpy( lpMsg->m_szStr[2], szIp );
            lpMsg->m_nData[2] = usPort;
        }

        if( 0 == nRet )
        {
            char szIp[100] = { 0 };
            xos_u16 usPort = 0;
            //lpSocket->get_socket_addr( szIp, &usPort );
            strcpy( lpMsg->m_szStr[1], szIp );
            lpMsg->m_nData[1] = usPort;
        }

        if( 0 == nRet )
        {
            lpRetKey->m_bCanWrite = true;
            lpRetKey->m_pHandle = lpAcceptTcp;
            lpMsg->m_lpData[2] = lpRetKey;
            lpMsg->m_bData[0] = true;
            lpRetKey->m_nEpollFlags = EPOLLIN | EPOLLOUT | EPOLLERR;
            lpRetKey->m_enumHandleType = aio_key::H_TYPE_TCP_NORM;
            lpRetKey->m_eState = aio_key::KEY_STATE_RUNNING;
            lpAcceptTcp->m_socket.set_async( true );

            struct epoll_event ev;
            ev.events = lpRetKey->m_nEpollFlags;
            ev.data.ptr = lpRetKey;
            epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_ADD, lpAcceptTcp->m_socket.m_hSocket, &ev );
        }

        if( 0 == nRet )
        {
            if( lpList )
            {
                lpList->push_back( lpMsg );
            }
            else
            {
                m_pAioImpl->m_pCallbackThread->notify( lpMsg );
            }

            lpAcceptTcp = 0;
            lpRetKey = 0;
            lpMsg = 0;
        }
        else
        {
            lpKey->m_pend_accept_list.push_front( lpMsg );
            lpMsg = 0;
        }

        if( lpAcceptTcp )
        {
            lpAcceptTcp->release();
            lpAcceptTcp = 0;
        }

        if( lpRetKey )
        {
            lpRetKey->m_pHandle = 0;
            lpRetKey->release();
            lpRetKey = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : recv len
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_recv( aio_key * lpKey )
    {
        int nRet = 0;

        tcp_impl * lpTcp = ( tcp_impl * )lpKey->m_pHandle;
        socket_impl * lpSocket = &lpTcp->m_socket;
        msg_impl::LIST * lpList = 0;
        msg_impl * lpMsg = 0;
        buf_impl * lpBuf = 0;
        int nRecvLen = 0;

        if( 0 == lpKey->m_pend_recv_list.size() )
        {
            g_pXosImpl->create( i_xos::XOS_OBJ_BUF, ( void ** )&lpBuf );
            g_pXosImpl->create( i_xos::XOS_OBJ_MSG, ( void ** )&lpMsg );
            lpList = &lpKey->m_recv_list;
            lpMsg->m_lpData[1] = lpKey;
            lpMsg->m_lpBuf[0] = lpBuf;
        }
        else
        {
            lpMsg = lpKey->m_pend_recv_list.front();
            lpKey->m_pend_recv_list.pop_front();
            lpBuf = lpMsg->m_lpBuf[0];
        }

        lpMsg->m_nData[0] = i_aio::AIO_TCP_RECV_RET;
        lpMsg->m_bData[0] = false;

        lpBuf->m_nLen = lpSocket->tcp_recv( lpBuf->m_szBuf, sizeof( lpBuf->m_szBuf ) );

        if( lpBuf->m_nLen > 0 )
        {
            lpMsg->m_nData[1] = lpBuf->m_nLen;
            lpMsg->m_bData[0] = true;
        }
        else
        {
            lpMsg->m_bData[0] = false;
            lpMsg->m_nData[1] = 0;
            lpBuf->m_nLen = 0;
        }

        nRecvLen = lpBuf->m_nLen;

        if( !lpList )
        {
            m_pAioImpl->m_pCallbackThread->notify( lpMsg );
        }
        else
        {
            lpList->push_back( lpMsg );
        }

        if( nRecvLen <= 0 )
        {
            epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_DEL, lpSocket->m_hSocket, 0 );
            lpKey->finish_request( m_pAioImpl );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : send bytes
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_send( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        tcp_impl * lpTcp = ( tcp_impl * )lpKey->m_pHandle;
        socket_impl * lpSocket = &lpTcp->m_socket;
        msg_impl * lpMsg = 0;
        buf_impl * lpBuf = 0;
        int nLastFlag = lpKey->m_nEpollFlags;

        if( 0 == nRet )
        {
            if( !lpKey->m_bCanWrite )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            lpMsg = lpKey->m_pend_send_list.front();

            if( lpMsg )
            {
                lpKey->m_pend_send_list.pop_front();
                lpBuf = lpMsg->m_lpBuf[0];
            }
            else
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            int ret = lpSocket->tcp_send( lpBuf->m_szBuf + lpBuf->m_nPos, lpBuf->m_nLen );
            bool bSave = true;

            if( ret > 0 )
            {
                lpMsg->m_nData[1] += ret;
                lpBuf->m_nPos += ret;
                lpBuf->m_nLen -= ret;

                if( 0 == lpBuf->m_nLen )
                {
                    lpMsg->m_bData[0] = true;
                    bSave = false;
                }
            }
            else
            {
                nRet = 1;
            }

            if( bSave )
            {
                lpKey->m_pend_send_list.push_front( lpMsg );
                lpKey->m_bCanWrite = false;
                lpMsg = 0;
            }
        }

        if( lpMsg )
        {
            lpBuf->m_nLen = lpMsg->m_nData[1];
            lpBuf->m_nPos = lpBuf->m_nPos - lpBuf->m_nLen;
            m_pAioImpl->m_pCallbackThread->notify( lpMsg );
            lpMsg = 0;
        }

        if( lpKey->m_pend_send_list.front() )
        {
            lpKey->m_nEpollFlags |= EPOLLOUT;
        }
        else
        {
            lpKey->m_nEpollFlags &= ~EPOLLOUT;
        }

        if( nLastFlag != lpKey->m_nEpollFlags )
        {
            struct epoll_event ev;
            ev.data.ptr = lpKey;
            ev.events = lpKey->m_nEpollFlags;
            epoll_ctl( m_pAioImpl->m_hEpoll, EPOLL_CTL_MOD, lpSocket->m_hSocket, &ev );
        }

        return nRet;
    }

} // xos
